package com.bang.study.lock;

import org.elasticsearch.action.*;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.InetSocketTransportAddress;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentParser;
import org.elasticsearch.index.engine.VersionConflictEngineException;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.transport.client.PreBuiltTransportClient;

import java.io.IOException;
import java.net.InetAddress;
import java.util.HashMap;
import java.util.Map;

/**
 * @ClassName: FsLockDemo
 * @author: BANG
 * @Date: 2020/01/16 13:40
 * @Version: 1.0
 * @Modified By:
 * @Desc: 基于version的乐观锁并发控制
 * <p>
 * 全局锁的上锁实验
 */
public class FsDocumentLock {

    /**
     * POST /fs/lock/1/_update
     * {
     * "upsert": { "process_id": 123 },
     * "script": {
     * "lang": "groovy",
     * "file": "judge-lock",
     * "params": {
     * "process_id": 123
     * }
     * }
     * }
     *
     * @param args
     * @throws IOException
     */
    public static void main(String[] args) throws IOException {
        Settings settings = Settings.builder()
                .put("cluster.name", "elasticsearch")
                .put("client.transport.sniff", true)
                .build();

        TransportClient client = new PreBuiltTransportClient(settings).addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));

        try {
            Map<String, String> options = new HashMap<String, String>();
            Map<String, Object> params = new HashMap<String, Object>();
            params.put("process_id", 123);
            UpdateResponse updateResponse = client.prepareUpdate("fs", "lock", "1")
                    .setUpsert(XContentFactory.jsonBuilder().startObject().field("process_id", 123).endObject())
                    .setScriptedUpsert(true)
                    .setScript(new Script(ScriptType.FILE, "groovy", "judge-lock", options, params))
                    .get();

            IndexRequest indexRequest = new IndexRequest("airobot", "user", "1")
                    .source(
                            XContentFactory.jsonBuilder()
                                    .startObject()
                                    .field("name", "张三")
                                    .field("age", 20)
                                    .endObject()
                    );
            client.index(indexRequest);

            GetRequest getRequest = new GetRequest("fs").refresh(true);
            ActionFuture<GetResponse> getResponseActionFuture = client.get(getRequest);


            SearchResponse searchResponse = client.prepareSearch("fs").setTypes("lock")
                    .setQuery(QueryBuilders.termQuery("process_id", 123))
                    .setScroll(new TimeValue(60000))
                    .get();
            BulkRequest bulkRequest = new BulkRequest();

            while (searchResponse.getHits().getHits().length != 0) {
                for (SearchHit hit : searchResponse.getHits().getHits()) {
                    String id = hit.getId();
                    System.err.println(id);
                    bulkRequest.add(new DeleteRequest("fs", "lock", id));
                }
                searchResponse = client.prepareSearchScroll(searchResponse.getScrollId())
                        .setScroll(new TimeValue(60000))
                        .execute()
                        .actionGet();
            }

            client.bulk(bulkRequest);

        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("Exception");
        } finally {
            client.close();
        }

    }
}
